package com.rrd.dw.utils;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HBaseUtil {
    private static final Logger log = LoggerFactory.getLogger(HBaseUtil.class);

    public static Configuration getConfiguration() {
        Configuration conf = HBaseConfiguration.create();

        conf.set("hbase.zookeeper.quorum", "datanode1,datanode2,datanode3");
        // conf.set("hbase.master", "hadoop-1");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        log.info("hbase.master:" + conf.get("hbase.master"));
        log.info(
            "hbase.zookeeper.quorum:" + conf.get("hbase.zookeeper.quorum"));
        return conf;

    }

    /**
     * Returns a HBaseAdmin instance. This instance is shared between
     * HBaseTestingUtility instance users. Don't close it, it will be closed
     * automatically when the cluster shutdowns
     * 
     * @return The HBaseAdmin instance.
     * @throws IOException
     */
    public static synchronized Admin getHBaseAdmin() throws IOException {
        Connection connection = ConnectionFactory
            .createConnection(getConfiguration());
        Admin admin = connection.getAdmin();
        return admin;
    }

    public synchronized static Connection getConnection() throws IOException {
        Connection connection = ConnectionFactory
            .createConnection(getConfiguration());

        return connection;
    }

    public synchronized static Connection getConnection(Configuration conf) {
        Connection connection = null;
        try {
            connection = ConnectionFactory.createConnection(conf);

        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return connection;
    }

    /**
     * 根据表名通过连接池获取Htable对象
     * 
     * @param tableName
     *        hbase表名称
     * @return
     * @throws IOException
     */
    public static Table getHTable(String tableName) throws IOException {
        return getConnection().getTable(TableName.valueOf(tableName));
    }

    public static void closeTable(Table htable) {
        try {
            if (htable != null) {
                htable.close();
            }

        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    public static final byte[] INF_FAMILY = Bytes.toBytes("inf");
    public static final byte[] JOB_FAMILY = Bytes.toBytes("job");

    public static void scanLimit(String tableName) throws IOException {
        Table ht = getHTable(tableName);
        Scan scan = new Scan();
        scan.addColumn(INF_FAMILY, Bytes.toBytes("card_bill_report"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("system_id"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("start_time"));
        scan.addColumn(JOB_FAMILY, Bytes.toBytes("job_id"));
        Long startTime = DateUtil.formateToTimestamp("2017-09-12");
        Long endTime = DateUtil.formateToTimestamp("2017-09-13");
        scan.setTimeRange(startTime, endTime + 14400000);
        ResultScanner rs = ht.getScanner(scan);
        String family = "";
        int i = 1;
        for (Result r : rs) {
            Object[] keyarr = decodeRowkey(r.getRow());
            if (!keyarr[1].toString().equals("45022119920407291x")) {
                continue;
            }
            System.out
                .println("rowkey :" + keyarr[0].toString() + "," + keyarr[1]);
            NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = r
                .getMap();
            Set<byte[]> mapSet = map.keySet();

            if (i <= 0) {
                break;
            }
            for (Iterator<byte[]> mapIt = mapSet.iterator(); mapIt.hasNext();) {
                family = Bytes.toString(mapIt.next());
                // if(!family.equals("inf")){
                // continue;
                // }
                NavigableMap<byte[], NavigableMap<Long, byte[]>> familyMap = map
                    .get(Bytes.toBytes(family));
                Set<byte[]> familyMapSet = familyMap.keySet();
                for (Iterator<byte[]> familyIt = familyMapSet
                    .iterator(); familyIt.hasNext();) {
                    byte[] keyColumn = (byte[]) familyIt.next();
                    NavigableMap<Long, byte[]> columnMap = familyMap
                        .get(keyColumn);
                    Map<Long, String> mapColumn = new HashMap<Long, String>();
                    Set<Long> columnMapSet = columnMap.keySet();
                    for (Iterator<Long> columnIt = columnMapSet
                        .iterator(); columnIt.hasNext();) {
                        Long columnKey = (Long) columnIt.next();
                        String columnValue = new String(
                            columnMap.get(columnKey));
                        System.out.println(family + ":"
                            + Bytes.toString(keyColumn) + "=" + columnValue);
                        if (columnValue != null && !columnValue.equals("")) {
                            break;
                        }

                    }
                }
            }
            i--;
        }
    }

    public static void getByRowKey(String tableName, byte[] rowkey)
            throws IOException, ParseException {
        SimpleDateFormat formatter = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
        Long startTime = HBaseUtil.parseDate("2018-02-09").getTime();
        Long endTime = HBaseUtil.parseDate("2018-02-10").getTime();
        Table table = getHTable(tableName);
        Get g = new Get(rowkey);
        g.setTimeRange(startTime,endTime);
        Result result = table.get(g);
        List<Cell> cellList = result.listCells();
        for (Cell cell : cellList) {
            System.out.println(Bytes.toString(cell.getFamily()) + "-"
                + Bytes.toString(cell.getQualifier()) + "-"
                + Bytes.toString(cell.getValue()) + "-"
                + formatter.format(new Date(cell.getTimestamp())));
        }
    }

    public static HTableDescriptor getHTableDescriptor(String tablename)
            throws IOException {
        Table table = getHTable(tablename);

        HTableDescriptor htable_desc = table.getTableDescriptor();

        return htable_desc;
    }

    public static final int SIZEOF_MD5 = 16;
    public static final int SIZEOF_USERKEY = 32;

    public static Object[] decodeRowkey(byte[] raw) {
        if (raw.length < SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
            return null;
        }
        Object[] ret = new Object[2];
        ret[0] = Long.MAX_VALUE
            - Bytes.toLong(raw, SIZEOF_MD5, Bytes.SIZEOF_LONG);

        if (raw.length > SIZEOF_MD5 + Bytes.SIZEOF_LONG) {
            ret[1] = Bytes.toString(raw, SIZEOF_MD5 + Bytes.SIZEOF_LONG);
        }

        return ret;
    }

    public static byte[] encodeUserRowkey(String userKey, long timestamp) {
        if (userKey == null) {
            return null;
        }
        return ArrayUtils.addAll(Bytes.toBytes(userKey),
            Bytes.toBytes(Long.MAX_VALUE - timestamp));
    }

    /**
     * @param raw
     * @return [0] = userKey
     *         [1] = timestamp
     */
    public static Object[] decodeUserRowkey(byte[] raw) {
        if (raw.length < HBaseUtil.SIZEOF_USERKEY + Bytes.SIZEOF_LONG) {
            return null;
        }
        Object[] ret = new Object[2];
        ret[0] = Bytes.toString(raw, 0, HBaseUtil.SIZEOF_USERKEY);
        ret[1] = Long.MAX_VALUE - Bytes.toLong(raw, HBaseUtil.SIZEOF_USERKEY);
        return ret;
    }

    /**
     * @param raw
     * @return [0] = mobile
     *         [1] = timestamp
     */
    public static Object[] decodeMobileRowkey(byte[] raw) {
        if (raw.length < Bytes.SIZEOF_LONG) {
            return null;
        }

        Object[] ret = new Object[2];
        ret[0] = StringUtils
            .reverse(Bytes.toString(raw, 0, raw.length - Bytes.SIZEOF_LONG));
        ret[1] = Long.MAX_VALUE
            - Bytes.toLong(raw, raw.length - Bytes.SIZEOF_LONG);

        return ret;
    }

    public static Date parseDate(String strDate) throws ParseException {
        Date result = null;
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        result = sdf.parse(strDate);
        return result;
    }

    public static String getStatDimStr(String dim, long timestamp) {
        SimpleDateFormat formatter = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
        Date date = new Date(timestamp);
        String dateString = formatter.format(date);
        if ("H".equals(dim)) {
            return dateString.substring(0, 13);
        } else if ("D".equals(dim)) {
            return dateString.substring(0, 10);
        } else if ("W".equals(dim)) {
            Calendar rightNow = Calendar.getInstance();
            rightNow.setTime(date);
            return rightNow.get(Calendar.YEAR) + "-"
                + rightNow.get(Calendar.WEEK_OF_YEAR);
        } else if ("M".equals(dim)) {
            return dateString.substring(0, 7);
        }
        return "";
    }
    public static Object getValue(byte[] value) {
    	return getValue("string",value);
    }
    public static Object getValue(String type, byte[] value) {
        SimpleDateFormat formatter = new SimpleDateFormat(
            "yyyy-MM-dd HH:mm:ss");
        switch (type) {
            case "string":
                return value == null ? "" : Bytes.toString(value);
            case "date":
                return value == null ? new Date(0)
                    : formatter.format(new Date(Bytes.toLong(value)));
            case "bigdecimal":
                return value == null ? new BigDecimal(0)
                    : Bytes.toBigDecimal(value);
            case "boolean":
                return value != null && Bytes.toBoolean(value);
            case "double":
                return value == null ? 0 : Bytes.toDouble(value);
            case "float":
                return value == null ? 0 : Bytes.toFloat(value);
            case "int":
                return value == null ? 0 : Bytes.toInt(value);
            case "short":
                return value == null ? 0 : Bytes.toShort(value);
            case "long":
                return value == null ? 0 : Bytes.toLong(value);
            default:
                return Bytes.toString(value);

        }
    }

    public static void main(String[] args) throws IOException {
        try {
            Long startTime = HBaseUtil.parseDate("2018-02-09").getTime();
            Long endTime = HBaseUtil.parseDate("2018-02-10").getTime();
            System.out.println(startTime +"-"+endTime);
        } catch (ParseException e) {
            e.printStackTrace();
        }


    }

}
